package mobao.log.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.istack.internal.Nullable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Date;

/**
 * Created by yanghongquan on 2018/9/13 15:41
 */
@Component
public class Provider {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    private static ObjectMapper objectMapper = new ObjectMapper();

    public void send(String key,String msg,Object data) {
        Message message = new Message();
        message.setId("MBK_"+System.currentTimeMillis());
        message.setMsg(msg);
        message.setData(data);
        message.setSendTime(new Date());
        String sendMsg = "";
        try {
            sendMsg = objectMapper.writeValueAsString(message);
        } catch (JsonProcessingException e) {
            //省略异常处理内容
            e.printStackTrace();
        }
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(key,sendMsg);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("MOBO消息发送异常:" + throwable.getMessage());
            }

            @Override
            public void onSuccess(@Nullable SendResult<Integer, String> integerStringSendResult) {
                System.out.println("MOBO消息发送成功:"+integerStringSendResult.toString());
            }
        });
    }

}
